package com.xyzwps.reactor;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class Ex03Scheduler {
    public static void main(String[] args) throws IOException {
        System.out.println("选择 demo：");
        var reader = new BufferedReader(new InputStreamReader(System.in));
        switch (reader.readLine()) {
            case "pubOn" -> demoPublishOn();
            case "subOn" -> demoSubscribeOn();
            default -> System.out.println("啥也没选");
        }
    }

    /**
     * publishOn 用于在订阅链中间上下文（scheduler）
     *
     * publishOn 改变调用处之后的订阅链上下文
     */
    static void demoPublishOn() {
        var scheduler = Schedulers.newParallel("parallel-scheduler", 4);

        final Flux<String> flux = Flux
                .range(1, 4)
                .map(i -> {
                    System.out.println("upstream thread " + Thread.currentThread().getName());
                    return 10 + i;
                })
                .publishOn(scheduler)
                .map(i -> {
                    System.out.println("downstream thread " + Thread.currentThread().getName());
                    return "value " + i;
                });

        new Thread(() -> {
            System.out.println("launch thread " + Thread.currentThread().getName());
            flux.subscribe(System.out::println);
        }).start();
    }

    /**
     * subscribeOn 改变调用处之前的订阅链上下文
     *
     * 这个比较符合直觉：订阅链前面的执行发生在订阅的线程中，这里指定订阅使用的 scheduler，所以是改变之前的上下文
     */
    static void demoSubscribeOn() {
        var scheduler = Schedulers.newParallel("parallel-scheduler", 4);

        final Flux<String> flux = Flux
                .range(1, 4)
                .map(i -> {
                    System.out.println("upstream thread " + Thread.currentThread().getName());
                    return 10 + i;
                })
                .subscribeOn(scheduler)
                .map(i -> {
                    System.out.println("downstream thread " + Thread.currentThread().getName());
                    return "value " + i;
                });

        new Thread(() -> {
            System.out.println("launch thread " + Thread.currentThread().getName());
            flux.subscribe(System.out::println);
        }).start();
    }
}
